package cc.ifok.hub;

import cc.ifok.kit.Command;
import cc.ifok.kit.Message;
import cc.ifok.kit.Strings;
import cc.ifok.lpc.LPCClient;
import cc.ifok.rpc.RPCClient;

import java.util.UUID;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Supplier;

public final class Exchange { private Exchange() {}
    private static final ScheduledExecutorService LPC_STATER = Executors.newSingleThreadScheduledExecutor();
    private static final ExecutorService RPC_WORKER = Executors.newSingleThreadExecutor();
    private static final ExecutorService LPC_WORKER = Executors.newSingleThreadExecutor();
    private static final AtomicReference<RPCClient> RPC = new AtomicReference<>();
    private static final AtomicReference<LPCClient> LPC = new AtomicReference<>();

    public static void run(RPCClient rpc, LPCClient lpc) {RPC.set(rpc); LPC.set(lpc);
        LPC_WORKER.submit(()->{while(Message.working()){try5s(()->RPC.get().send(Message.lps().stringify()));}});
        LPC_STATER.scheduleAtFixedRate(() -> RPC.get().send(states()), 128, 5120, TimeUnit.MILLISECONDS);
        RPC_WORKER.submit(()->{while(Message.working()){Message message = Message.rps();
            if (Command.SHELLED.name().equalsIgnoreCase(message.cmd())) {
                if (!try5s(() -> LPC.get().send(message.sid(), message.val()))) {//提示Remote发送失败
                    RPC.get().send(new Message(message.sid(), Command.ALERTED, Strings.EMPTY).stringify());
                }
            } else if (Command.CLOSED.name().equalsIgnoreCase(message.cmd())) {
                LPC.get().close(message.sid());
            } else if (Command.UPLOAD.name().equalsIgnoreCase(message.cmd())) {
                upload(message.sid(), message.val());
            } else if (Command.DOWNLOAD.name().equalsIgnoreCase(message.cmd())) {
                download(message.sid(), message.val());
            }
        }});
    }

    private static boolean try5s(Supplier<Boolean> handler) {
        final AtomicInteger counter = new AtomicInteger(1);
        final AtomicBoolean result = new AtomicBoolean();
        do {// 消息发送失败则重试times次
            if(handler.get()) {
                result.set(true); break;
            } else {
                if (counter.get() < 5) {
                    try {
                        TimeUnit.SECONDS.sleep(counter.get());
                    } catch(Throwable e){/*ignore exception*/}
                }
            }
        } while(counter.getAndIncrement()<5);return result.get();
    }
    private static String states() {// VPC状态数据
        //TODO states
        String values = Strings.EMPTY + LPC.get().counts();
        return new Message(UUID.randomUUID().toString(), Command.STATED, values).stringify();
    }
    private static void upload(String sid, String value) {// 上传文件
        //TODO upload
    }
    private static void download(String sid, String value) {// 下载文件
        //TODO download
    }
}
